/*******************************************************************************
 * Copyright (c) 2012, 2017 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *   http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial contribution
 *******************************************************************************/


#include "project.h"


//接收内存
void *recv_neicun_thread(void *arg)
{
    int id = *((int *)arg);
    free(arg);  // 参数由主线程动态分配，线程内部释放

    if (id < 0 || id > 1) {
        printf("接收者编号只能是 0 或 1\n");
        pthread_exit(NULL);
    }

    int shmid = shmget(SHM_KEY, sizeof(struct shmseg), 0666);
    if (shmid == -1) {
        perror("shmget");
        pthread_exit(NULL);
    }

    int semid = semget(SEM_KEY, 1, 0666);
    if (semid == -1) {
        perror("semget");
        pthread_exit(NULL);
    }

    struct shmseg *shm = (struct shmseg *)shmat(shmid, NULL, 0);
    if (shm == (void *)-1) {
        perror("shmat");
        pthread_exit(NULL);
    }

    while (running) {
        char local_buf[SHM_SIZE] = {0};

        sem_wait(semid);

        if (shm->written == 1 && shm->read_flags[id] == 0) {
            strncpy(local_buf, shm->text, SHM_SIZE);
            shm->read_flags[id] = 1;

            if (shm->read_flags[0] == 1 && shm->read_flags[1] == 1) {
                shm->written = 0;
            }
        }

        sem_signal(semid);

        if (local_buf[0] != '\0') {
            printf("接收者%d 收到消息: %s\n", id, local_buf);

            pthread_mutex_lock(&global_send_mutex);
            strncpy(global_send_buffer, local_buf, SHM_SIZE);
            pthread_mutex_unlock(&global_send_mutex);
        }

        usleep(100000); // 100ms 轮询（比 sleep(1) 更及时）
    }

    shmdt(shm);  // 清理映射
    pthread_exit(NULL);
}


//发送队列信息函数
void *send_duilie_thread(void *arg)
{
    int msgid = msgget(MSG_KEY, 0666 | IPC_CREAT);
    if (msgid == -1) {
        perror("msgget failed");
        pthread_exit(NULL);
    }

    struct msgbuf msg;
    msg.mtype = 1;

    while (running)
    {
        pthread_mutex_lock(&global_recv_mutex);

        if (strlen(global_recv_buffer) > 0) {
            strncpy(msg.mtext, global_recv_buffer, sizeof(msg.mtext) - 1);
            global_recv_buffer[0] = '\0';  // 清空，防止重复发送

            pthread_mutex_unlock(&global_recv_mutex);

            if (msgsnd(msgid, &msg, strlen(msg.mtext) + 1, 0) == -1) {
                perror("msgsnd failed");
                break;
            }

            printf("已通过消息队列发送: %s\n", msg.mtext);
        } else {
            pthread_mutex_unlock(&global_recv_mutex);
            usleep(100 * 1000);  // 100ms 避免忙等
        }
    }

    pthread_exit(NULL);
}


// 接收线程函数（主动拉取消息）mqtt
void *receive_thread(void *arg)
{
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n",
           TOPIC_recv, CLIENTID, QOS);

    while (running)
    {
        MQTTClient_message *msg = NULL;
        char *topic = NULL;
        int rc;

        // 🔒 锁定 client 接收消息
        pthread_mutex_lock(&client_mutex);
        rc = MQTTClient_receive(client, &topic, NULL, &msg, 1000); // 阻塞1秒
        pthread_mutex_unlock(&client_mutex);

        // ✅ 接收成功 且 msg 不为空
        if (rc == MQTTCLIENT_SUCCESS && msg != NULL)
        {
            printf("Received on [%s]: %.*s\n",
                   topic, msg->payloadlen, (char *)msg->payload);

            // 🔒 锁定共享缓冲区，复制数据
            pthread_mutex_lock(&global_recv_mutex);
            memset(global_recv_buffer, 0, sizeof(global_recv_buffer));  // 可选，防残留
            strncpy(global_recv_buffer, msg->payload, sizeof(global_recv_buffer) - 1); // 留空间给 '\0'
            pthread_mutex_unlock(&global_recv_mutex);

            // ✅ 清理资源
            MQTTClient_free(topic);
            MQTTClient_freeMessage(&msg);
        }
        else
        {
            // rc 不为 SUCCESS 或没有 msg，不做任何操作
            if (msg != NULL) {
                MQTTClient_freeMessage(&msg);
            }
            if (topic != NULL) {
                MQTTClient_free(topic);
            }
        }

        // 可以加一个短 sleep 避免 CPU 空转（可选）
        usleep(100 * 1000); // 100ms
    }

    pthread_exit(NULL);
}

void mq_send()
{
     if (strlen(global_send_buffer) > 0) {
        MQTTClient_message pubmsg = MQTTClient_message_initializer;
        pubmsg.payload = global_send_buffer;
        pubmsg.payloadlen = (int)strlen(global_send_buffer);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;

        MQTTClient_deliveryToken token;
        int rc = MQTTClient_publishMessage(client, TOPIC_send, &pubmsg, &token);
        if (rc != MQTTCLIENT_SUCCESS) {
            printf("Publish failed: %d\n", rc);
        } else {
            // 清空 buffer，防止重复发
            global_send_buffer[0] = '\0';
        }
    }
}

// 发送线程函数mqtt
void *send_thread(void *arg)
{
   while (running) {
    pthread_mutex_lock(&global_send_mutex);
    pthread_mutex_lock(&client_mutex);
    if(mode==1)
    {
        mq_send();
    }
    if(mode==2)//对比
    {
       // char global_mode2_buffer[512] = {0};//mode2的中转
       //char global_send_buffer[512] = {0};//内存到mq
       if(strcmp(global_mode2_buffer,global_send_buffer)==0)
       {
        pthread_mutex_unlock(&global_send_mutex);
        pthread_mutex_unlock(&client_mutex);
            continue;
       }
       else
       {
       mq_send();
       }
    }
    if(mode==3)//刷新
    {
        if(command==1)
        {
            mq_send();
        }
        else
        {
            pthread_mutex_unlock(&global_send_mutex);
            pthread_mutex_unlock(&client_mutex);
            continue;
        }
    }
    pthread_mutex_unlock(&global_send_mutex);
    pthread_mutex_unlock(&client_mutex);
    sleep(1);  // 控制发送频率
}


   
}

int main(int argc, char *argv[])
{
    int rc = 0;
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 订阅线程//订阅的是mqtt主题
    pthread_mutex_lock(&client_mutex);
    MQTTClient_subscribe(client, TOPIC_recv, QOS);
    pthread_mutex_unlock(&client_mutex);
    //


    //pthread_t recv_neicun;
int *receiver_id = malloc(sizeof(int));//内存线程传入参数
*receiver_id = 0;  // 或 1
//pthread_create(&recv_neicun, NULL, recv_neicun_thread, receiver_id);

    //

    // 创建线程
    pthread_t recv_tid, send_tid,send_duilie,recv_neicun;
    pthread_create(&recv_tid, NULL, receive_thread, NULL);
    pthread_create(&send_tid, NULL, send_thread, NULL);
    pthread_create(&send_duilie, NULL, send_duilie_thread, NULL);
    pthread_create(&recv_neicun, NULL, recv_neicun_thread, receiver_id);
    // 主线程控制退出
    // 不再 getchar()，由输入 exit 控制退出
while (running) {
    sleep(1);
}
  
    
    
    pthread_join(send_tid, NULL);
    pthread_join(recv_tid, NULL);
    pthread_join(send_duilie,NULL);
    pthread_join(recv_neicun,NULL);
    MQTTClient_unsubscribe(client, TOPIC_recv);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    pthread_mutex_destroy(&client_mutex);
    pthread_mutex_destroy(&global_send_mutex);
    pthread_mutex_destroy(&global_recv_mutex);
    return 0;
}
